#include "RPCActor.h"
#include "RPCManager.h"
#include "NetBuffer.h"

RPCActor::RPCActor()
: mgr_(nullptr)
, is_ready_(false)
{
}

RPCActor::~RPCActor()
{
}

void RPCActor::RPCInvoke(const INetPacket &pck,
        const std::function<void(INetStream&, int32, bool)> &cb,
        AsyncTaskOwner *owner, time_t timeout)
{
    RPCManager *mgr = GetCacheRPCManager();
    RPCTransInvoke(mgr->null_packet_, pck, cb, owner, timeout);
}

void RPCActor::RPCTransInvoke(const INetPacket &trans, const INetPacket &pck,
        const std::function<void(INetStream&, int32, bool)> &cb,
        AsyncTaskOwner *owner, time_t timeout)
{
    RPCManager *mgr = GetCacheRPCManager();
    uint64 requestSN = mgr->request_sn_.fetch_add(1);
    NetBuffer args(requestSN);

    bool isCopyPck = false;
    auto requestInfoPtr = std::make_shared<RPCManager::RequestInfo>();
    auto &requestInfo = *requestInfoPtr;
    if (is_ready_) {
        requestInfo.trans = nullptr;
        requestInfo.pck = nullptr;
    } else {
        requestInfo.trans = !trans.IsOpNil() ? pck.Clone() : nullptr;
        requestInfo.pck = pck.Clone();
        requestInfo.args = args.CastBufferString();
        isCopyPck = true;
    }
    if (cb) {
        requestInfo.cb = cb;
        if (owner != nullptr) {
            requestInfo.owner = owner->linked_from_this();
        } else {
            requestInfo.owner = mgr->null_owner_;
        }
    }

    requestInfo.timeout = timeout;
    mgr->PutRequestInfo(this, requestSN, std::move(requestInfoPtr));

    if (is_ready_) {
        PushRPCPacket(trans, pck, args.CastBufferStringView());
    } else if (!isCopyPck) {
        mgr->DoReply(nullptr, { requestSN, RPCErrorCancel, true });
    }
}

void RPCActor::RPCReply(const INetPacket &pck, uint64 sn, int32 err, bool eof)
{
    RPCManager *mgr = GetCacheRPCManager();
    RPCTransReply(mgr->null_packet_, pck, sn, err, eof);
}

void RPCActor::RPCTransReply(const INetPacket &trans, const INetPacket &pck,
        uint64 sn, int32 err, bool eof)
{
    NetBuffer args(sn, err, eof);
    PushRPCPacket(trans, pck, args.CastBufferStringView());
}

RPCActor::ReplyMetaInfo RPCActor::ReadReplyMetaInfo(INetPacket &pck)
{
    ReplyMetaInfo info;
    size_t infoLen = sizeof(info.sn) + sizeof(info.err) + sizeof(info.eof);
    if (pck.GetReadableSize() < infoLen) {
        THROW_EXCEPTION(NetStreamException());
    }

    size_t anchor = pck.GetReadPos();
    pck.AdjustReadPos(pck.GetReadableSize() - infoLen);
    pck >> info.sn >> info.err >> info.eof;
    pck.Shrink(pck.GetTotalSize() - infoLen);
    pck.AdjustReadPos(anchor - pck.GetReadPos());

    return info;
}

RPCActor::RequestMetaInfo RPCActor::ReadRequestMetaInfo(INetPacket &pck)
{
    RequestMetaInfo info;
    size_t infoLen = sizeof(info.sn);
    if (pck.GetReadableSize() < infoLen) {
        THROW_EXCEPTION(NetStreamException());
    }

    size_t anchor = pck.GetReadPos();
    pck.AdjustReadPos(pck.GetReadableSize() - infoLen);
    pck >> info.sn;
    pck.Shrink(pck.GetTotalSize() - infoLen);
    pck.AdjustReadPos(anchor - pck.GetReadPos());

    return info;
}
